package com.jjb.spider.mqtt.service;

import com.alibaba.fastjson.JSONObject;
import com.google.common.base.Strings;
import com.jjb.spider.mqtt.ctx.MqttChannelHandlerContext;
import com.jjb.spider.mqtt.model.DeviceTopSubscribeMessage;
import com.jjb.spider.mqtt.model.QueueMessage;
import com.jjb.spider.mqtt.model.TopSubscribeMessage;
import com.jjb.spider.mqtt.model.WillMeaasge;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.handler.codec.mqtt.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.ReactiveStringRedisTemplate;
import reactor.core.publisher.Mono;
import reactor.util.function.Tuples;

import java.time.Duration;
import java.util.*;
import java.util.stream.Collectors;

public abstract class IMqttMessageService {

    private Logger log = LoggerFactory.getLogger(IMqttMessageService.class);

    @Autowired
    private ReactiveStringRedisTemplate reactiveStringRedisTemplate;

    @Value("${mqtt.message.timeout: 60}")
    private Integer messageTimeout ;


    /**
     *
     * @param topic：发送的主题
     * @param isDup: 是否重复消息
     * @param byteBuf:消息内容
     * @param messageId:消息
     * @param mqttQoS:消息等级
     */
    protected MqttPublishMessage getMqttPublishMessage(String topic, boolean isDup, byte[] byteBuf, int messageId, MqttQoS mqttQoS){
        MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH,false, mqttQoS,false,0);
        MqttPublishVariableHeader mqttPublishVariableHeader = new MqttPublishVariableHeader(topic,messageId );
        MqttPublishMessage mqttPublishMessage = new MqttPublishMessage(mqttFixedHeader,mqttPublishVariableHeader, Unpooled.wrappedBuffer(byteBuf));
        return mqttPublishMessage;
    }

    /**
     * 保存远程设备监听
     * @return
     */
    protected Mono<Boolean> saveRemoteCacheSubscribe(String deviceId, List<TopSubscribeMessage> topicSet) {

        Mono<Boolean> result = Mono.just(true);
        for (TopSubscribeMessage message : topicSet) {
            result = result.flatMap(bool ->
                 this.reactiveStringRedisTemplate.opsForHash()
                        .put(deviceId, message.getTopic(), JSONObject.toJSONString(message))
            );
        }
        return result;
    }


    private String getRemoteSubscribeKey(String deviceId) {
        return deviceId+"_subscribe";
    }

    private  byte[] copyByteBuf(ByteBuf byteBuf) {
        try {
            byte[] bytes = new byte[byteBuf.readableBytes()];
            byteBuf.readBytes(bytes);
            return bytes;
        }finally {
            if( byteBuf != null ) {
                //释放内存。避免泄露
                byteBuf.release();
            }
        }
    }

    public Mono<Void> publish(MqttChannelHandlerContext ctx, MqttPublishMessage mqttMessage) {

        MqttFixedHeader mqttFixedHeader = mqttMessage.fixedHeader();
        MqttPublishVariableHeader mqttPublishVariableHeader = mqttMessage.variableHeader();
        ByteBuf payload = null;
        Mono<Void> result = null;


        byte[] content = copyByteBuf(mqttMessage.payload());
        int messageId = mqttPublishVariableHeader.messageId();
        MqttQoS qosLevel = mqttFixedHeader.qosLevel();
        //获取是否覆盖老的消息
        boolean isRetain = mqttFixedHeader.isRetain();
        boolean isDup = mqttFixedHeader.isDup();
        String topicName = mqttPublishVariableHeader.topicName();
        Integer packetId = mqttPublishVariableHeader.packetId();


        List<DeviceTopSubscribeMessage> deviceTopSubscribeMessages = getSendDeviceIdByTopic(topicName);
        List<DeviceTopSubscribeMessage> as = getSendDeviceIdByTopic("#");
        if( as != null) {
            deviceTopSubscribeMessages.addAll(as);
        }


        switch (qosLevel) {
            case AT_MOST_ONCE:

                if( ! (deviceTopSubscribeMessages == null || deviceTopSubscribeMessages.isEmpty())) {

                    MqttPublishMessage mqttPublishMessage = getMqttPublishMessage(topicName, false, content, messageId, MqttQoS.AT_MOST_ONCE);
                    Mono<Boolean> mono = Mono.just(true);
                    for (DeviceTopSubscribeMessage deviceTopSubscribeMessage : deviceTopSubscribeMessages) {
                        mono = mono.flatMap(bool ->
                                sendMsgByDeviceId(deviceTopSubscribeMessage.getDeviceId(), mqttPublishMessage)
                        );
                    }
                    result = mono.flatMap(bool -> Mono.empty());
                } else {
                    result = Mono.empty();
                }
                break;
            case AT_LEAST_ONCE:
                if ( deviceTopSubscribeMessages == null || deviceTopSubscribeMessages.isEmpty() ) {
                    result = Mono.create(sink -> {
                        MqttMessageIdVariableHeader mqttMessageIdVariableHeaderBack = MqttMessageIdVariableHeader.from(packetId);
                        MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.PUBACK,isDup, MqttQoS.FAILURE, isRetain, 0x02);
                        MqttPubAckMessage pubAck = new MqttPubAckMessage(mqttFixedHeaderBack, mqttMessageIdVariableHeaderBack);
                        ctx.writeAndFlush(pubAck)
                                .addListener(future -> {
                                    sink.success(future.isSuccess());
                                });

                    }).flatMap(bool -> Mono.empty());
                } else {
                    Mono<Boolean> mono = Mono.just(true);

                    for (DeviceTopSubscribeMessage  deviceTopSubscribeMessage : deviceTopSubscribeMessages) {
                        int finalQosLevel = Math.min(deviceTopSubscribeMessage.getQos(), qosLevel.value());
                        MqttPublishMessage mqttPublishMessage = getMqttPublishMessage(topicName, false, content, messageId, MqttQoS.valueOf(finalQosLevel));
                        mono = mono.flatMap(bool ->
                                sendMsgByDeviceId(deviceTopSubscribeMessage.getDeviceId(), mqttPublishMessage)
                                        .map(b -> b || bool)
                        );
                    }

                    result =mono.flatMap(bool -> {
                        MqttMessageIdVariableHeader mqttMessageIdVariableHeaderBack = MqttMessageIdVariableHeader.from(packetId);
                        MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.PUBACK, isDup, bool?MqttQoS.AT_MOST_ONCE:MqttQoS.FAILURE, isRetain, 0x02);
                        MqttPubAckMessage pubAck = new MqttPubAckMessage(mqttFixedHeaderBack, mqttMessageIdVariableHeaderBack);
                        return Mono.create(sink -> {
                            ctx.writeAndFlush(pubAck).addListener(future -> {
                                sink.success(future.isSuccess());
                            });
                        });
                    }).flatMap(bool -> Mono.empty());
                }; break;
            case EXACTLY_ONCE:

                //保存设备消息
                Mono<Boolean> mono1 = this.saveMessage(messageId, content, qosLevel, topicName, ctx.getDeviceId(), packetId, isDup, isRetain);

                for (DeviceTopSubscribeMessage deviceTopSubscribeMessage : deviceTopSubscribeMessages) {
                    int finalQosLevel = Math.min(deviceTopSubscribeMessage.getQos(), qosLevel.value());
                    MqttPublishMessage mqttPublishMessage = getMqttPublishMessage(topicName, false, content, messageId, MqttQoS.valueOf(finalQosLevel));
                    mono1 = mono1.flatMap(bool -> sendMsgByDeviceId(deviceTopSubscribeMessage.getDeviceId(), mqttPublishMessage));
                }
                result = mono1.flatMap(bool -> Mono.empty());
            case FAILURE:
                break;
        }


        return result;
    }



    private Mono<Boolean> saveMessage(int messageId, byte[] content, MqttQoS qosLevel, String topicName, String deviceId, Integer packetId, Boolean isDup, Boolean isRetain) {
        QueueMessage message = new QueueMessage();
        message.setMessageId(messageId);
        message.setQosLevel(qosLevel.value());
        message.setTopicName(topicName);
        message.setBytes(content);
        message.setDeviceId(deviceId);
        message.setPacketId(packetId);
        message.setDup(isDup);
        message.setRetain(isRetain);
        return this.reactiveStringRedisTemplate.opsForValue().set(messageId+"_message", JSONObject.toJSONString(message), Duration.ofSeconds(messageTimeout));
    }

    private Mono<QueueMessage> getMessage(int messageId) {
        return this.reactiveStringRedisTemplate.opsForValue().get(messageId+"_message")
                .flatMap(s -> {
                    if(Strings.isNullOrEmpty(s)) {
                        return Mono.empty();
                    } else {
                        QueueMessage queueMessage =  JSONObject.parseObject(s, QueueMessage.class);
                        return Mono.just(queueMessage);
                    }
                });
    }

    private Mono<Boolean> delMessage(int messageId) {
        return this.reactiveStringRedisTemplate.opsForValue().delete(messageId+"_message");
    }

    /**
     * 给指定设备发消息
     * @param deviceId：设备ID
     * @param obj： 消息内容
     * @return
     */
    public abstract Mono<Boolean> sendMsgByDeviceId(String deviceId, Object obj);

    public abstract List<DeviceTopSubscribeMessage> getSendDeviceIdByTopic(String topics);

    public Mono<Void>  unsubscribe(MqttChannelHandlerContext ctx, MqttUnsubscribeMessage mqttMessage) {
        final List<String> topics = mqttMessage.payload().topics();
        if ( topics == null || topics.isEmpty() ) {
            return Mono.empty();
        }

        return this.removeRemoteCacheSubscribe(ctx.getDeviceId(), topics)
                .map(bool -> {
                    ctx.unsubscribe(topics);
                    return bool;
                }).map(bool -> {
                    unsuback(ctx, mqttMessage);
                    return bool;
                }).flatMap(bool -> Mono.empty());
    }

    /**
     * 取消绑定回复消息
     * @param ctx
     * @param mqttMessage
     */

    private void unsuback(MqttChannelHandlerContext ctx, MqttUnsubscribeMessage mqttMessage) {
        MqttMessageIdVariableHeader messageIdVariableHeader = (MqttMessageIdVariableHeader) mqttMessage.variableHeader();
        //	构建返回报文	可变报头
        MqttMessageIdVariableHeader variableHeaderBack = MqttMessageIdVariableHeader.from(messageIdVariableHeader.messageId());
        //	构建返回报文	固定报头
        MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.UNSUBACK, false, MqttQoS.AT_MOST_ONCE, false, 2);
        //	构建返回报文	取消订阅确认
        MqttUnsubAckMessage unSubAck = new MqttUnsubAckMessage(mqttFixedHeaderBack,variableHeaderBack);
        log.info("back--"+unSubAck.toString());
        ctx.writeAndFlush(unSubAck);
    }

    public  Mono<Void>  subscribe(MqttChannelHandlerContext ctx, MqttSubscribeMessage mqttMessage) {
        //获取所有的订阅主体
        List<TopSubscribeMessage> topSubscribeMessages = mqttMessage.payload().topicSubscriptions().stream().map(mqttTopicSubscription ->{
                    String topic = mqttTopicSubscription.topicName().trim();
                    MqttQoS qos = mqttTopicSubscription.option().qos();
                    return new TopSubscribeMessage(topic, qos.value());
        }
        ).collect(Collectors.toList());



        return this.saveRemoteCacheSubscribe(ctx.getDeviceId(), topSubscribeMessages)
                .map(bool -> {
                    ctx.saveSubscribe(topSubscribeMessages);
                    return bool;
                }).map(bool -> {
                    this.suback(ctx, mqttMessage, topSubscribeMessages.size());
                    return bool;
                }).flatMap(bool -> Mono.empty());
    }


    /**
     * 订阅成功的回复消息
     * @param ctx
     * @param mqttMessage
     */
    private void suback(MqttChannelHandlerContext ctx, MqttSubscribeMessage mqttMessage, int topSize) {
        MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.SUBACK, false, MqttQoS.AT_LEAST_ONCE, false, 0);
        MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(mqttMessage.variableHeader().messageId());
        List<Integer> grantedQoSLevels = new ArrayList<>(topSize);
        for (int i = 0; i < topSize; i++) {
            grantedQoSLevels.add(mqttMessage.payload().topicSubscriptions().get(i).qualityOfService().value());
        }
        MqttSubAckPayload payload = new MqttSubAckPayload(grantedQoSLevels);
        MqttSubAckMessage mqttSubAckMessage = new MqttSubAckMessage(mqttFixedHeader, variableHeader, payload);
        ctx.writeAndFlush(mqttSubAckMessage);
    }

    public abstract Mono<Void> closeConnectHandler(MqttChannelHandlerContext ctx);

    public Mono<Void>  closeConnect(MqttChannelHandlerContext ctx) {
        return closeConnectHandler(ctx);
    }

    protected Mono<Boolean> removeRemoteCacheSubscribe(String deviceId, List<String> topics) {
        String[] topicArray = new String[topics.size()];
        topicArray = topics.toArray(topicArray);
        return this.reactiveStringRedisTemplate.opsForSet().remove(getRemoteSubscribeKey(deviceId), topicArray).map(l -> true);
    }

    public Mono<Boolean> pubback(MqttChannelHandlerContext ctx, MqttPubAckMessage mqttMessage) {
        MqttMessageIdVariableHeader mqttPublishVariableHeader = mqttMessage.variableHeader();
        int messageId = mqttMessage.variableHeader().messageId();

        return getMessage(messageId)
                .flatMap(message -> {
                    MqttMessageIdVariableHeader mqttMessageIdVariableHeaderBack = MqttMessageIdVariableHeader.from(message.getPacketId());
                    MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.PUBACK,message.getDup(), MqttQoS.AT_MOST_ONCE, message.getRetain(), 0x02);
                    MqttPubAckMessage pubAck = new MqttPubAckMessage(mqttFixedHeaderBack, mqttMessageIdVariableHeaderBack);
                    return sendMsgByDeviceId(message.getDeviceId(), pubAck);
                });
    }

    public Mono<Boolean> pubrec(MqttChannelHandlerContext mqttChannelHandlerContext, MqttMessage mqttMessage) {
        MqttPubReplyMessageVariableHeader mqttPublishVariableHeader = (MqttPubReplyMessageVariableHeader) mqttMessage.variableHeader();
        int messageId = mqttPublishVariableHeader.messageId();


        return getMessage(messageId)
                .flatMap(message -> {
                    MqttFixedHeader mqttResultFixedHeader = new MqttFixedHeader(MqttMessageType.PUBREC,false, MqttQoS.AT_LEAST_ONCE,false,0x02);
                    MqttMessageIdVariableHeader from = MqttMessageIdVariableHeader.from(messageId);
                    MqttPubAckMessage mqttPubAckMessage = new MqttPubAckMessage(mqttResultFixedHeader,from);
                    sendMsgByDeviceId(message.getDeviceId(), mqttPubAckMessage);
                    return Mono.empty();
                });


    }

    public Mono<Boolean> pubrel(MqttChannelHandlerContext mqttChannelHandlerContext, MqttMessage mqttMessage) {
        MqttPubReplyMessageVariableHeader mqttPublishVariableHeader = (MqttPubReplyMessageVariableHeader) mqttMessage.variableHeader();
        int messageId = mqttPublishVariableHeader.messageId();

        return getMessage(messageId)
                .flatMap(message -> {

                    Collection<String> deviceIds = message.getDeviceIds();

                    MqttFixedHeader mqttResultFixedHeader = new MqttFixedHeader(MqttMessageType.PUBREL,false, MqttQoS.AT_LEAST_ONCE,false,0x02);
                    MqttMessageIdVariableHeader from = MqttMessageIdVariableHeader.from(messageId);
                    MqttPubAckMessage mqttPubAckMessage = new MqttPubAckMessage(mqttResultFixedHeader,from);

                    deviceIds.forEach(device -> {
                        sendMsgByDeviceId(device, mqttPubAckMessage);
                    });
                    return Mono.empty();
                });
    }

    public Mono<Boolean> pubcomp(MqttChannelHandlerContext mqttChannelHandlerContext, MqttMessage mqttMessage) {
        MqttPubReplyMessageVariableHeader mqttPublishVariableHeader = (MqttPubReplyMessageVariableHeader) mqttMessage.variableHeader();
        int messageId = mqttPublishVariableHeader.messageId();

        return getMessage(messageId)
                .map(message -> {
                    MqttFixedHeader mqttResultFixedHeader = new MqttFixedHeader(MqttMessageType.PUBCOMP,false, MqttQoS.AT_MOST_ONCE,false,0x02);
                    MqttMessageIdVariableHeader from = MqttMessageIdVariableHeader.from(messageId);
                    MqttPubAckMessage mqttPubAckMessage = new MqttPubAckMessage(mqttResultFixedHeader,from);
                    sendMsgByDeviceId(message.getDeviceId(), mqttPubAckMessage);
                    return message;
                }).flatMap(message ->
                     this.delMessage(messageId)
                ).flatMap(bool -> Mono.empty());
    }



    public void publish(WillMeaasge willMeaasge) {
        List<DeviceTopSubscribeMessage> collection = getSendDeviceIdByTopic(willMeaasge.getTopic());
        if ( collection == null || collection.isEmpty() ) {
            return;
        }

        int messageId = (int) (Math.random() * 10000000);
        MqttPublishMessage mqttPublishMessage = null;
        switch ( willMeaasge.getWillQos() ) {
            case 0:
                mqttPublishMessage = getMqttPublishMessage(willMeaasge.getTopic(), false, willMeaasge.getWillMessage(), messageId, MqttQoS.AT_MOST_ONCE);
                break;
            case 1:
            case 2:
                mqttPublishMessage = getMqttPublishMessage(willMeaasge.getTopic(), false, willMeaasge.getWillMessage(), messageId, MqttQoS.valueOf(willMeaasge.getWillQos()));
                break;
        }

        if ( mqttPublishMessage != null) {
            final MqttPublishMessage msg = mqttPublishMessage ;
            Mono<Boolean> mono = Mono.just(true);
            /*for(String deviceId : collection ){
                mono = mono.flatMap(bool -> sendMsgByDeviceId(deviceId, msg));
            }*/
            mono.subscribe();
        }
    }
}
